package com.practice.gmall.realtime.app.dwd;

import com.practice.gmall.realtime.app.BaseSQLApp;
import com.practice.gmall.realtime.util.MyKafkaUtil;
import com.practice.gmall.realtime.util.MySQLUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * ClassName: DWD_02_TradeCartAdd
 * Package: com.practice.gmall.realtime.app.dwd
 * Description: 过滤出加购购物车的事实数据，并关联维表base_dic, 将关联好后的表写入到Kafka中；
 * MySQL(cart_info) -> Kafka(ods_db) -> Flink(本类) -> Kafka(dwd_cart_add)
 * MySQL(base_dic) -> Flink(本类，使用look-up表) -> 提供给cart_info关联
 * @Author lzy.ethan@qq.com
 * @Create ${DATE} ${TIME}
 */
public class DWD_02_TradeCartAdd extends BaseSQLApp {
    public static void main( String[] args) {
        new DWD_02_TradeCartAdd().init(3335,"DWD_02_TradeCartAdd",2);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
      
        // 1. build dynamic table from kafka named ods_db
        MyKafkaUtil.readTopicOdsDb(tableEnv,"DWD_02_TradeCartAdd");
        
        // 2. filter cart_info data which stand for increasing sku_num from ods_db
        Table cartAdd = tableEnv.sqlQuery("" +
                "SELECT " +
                "`data`['id'] id, " +
                "`data`['user_id'] user_id, " +
                "`data`['sku_id'] sku_id, " +
                "`data`['cart_price'] cart_price, " +
                "if(`old`['sku_num'] is null,`data`['sku_num'],cast( cast(`data`['sku_num'] AS int) - cast(`old`['sku_num'] AS int) as  STRING)) sku_num, " +
                "`data`['sku_name'] sku_name, " +
                "`data`['is_checked'] is_checked, " +
                "`data`['create_time'] create_time, " +
                "`data`['operate_time'] operate_time, " +
                "`data`['is_ordered'] is_ordered, " +
                "`data`['order_time'] order_time, " +
                "`data`['source_type'] source_type, " +
                "`data`['source_id'] source_id, " +
                " pt " +
                "FROM " +
                "ods_db " +
                "WHERE " +
                "`database` = 'gmall2022' " +
                "    AND `table` = 'cart_info' " +
                "    AND ( `type` = 'insert' " +
                "   or (`type` = 'update'  " +
                "    AND  " +
                "    `old`['sku_num'] is not null  " +
                "    AND  " +
                "    cast(`data`['sku_num'] AS int) > cast(`old`['sku_num'] AS int)))");
        
        tableEnv.createTemporaryView("cart_add",cartAdd);

        //tableEnv.toAppendStream(cartAdd, Row.class).print(">>>>>>>>");// test whether get the table, streamEnv is needed


        // 3. select lookup table base_dic using mysql-connector
        tableEnv.executeSql(MySQLUtil.getBaseDicLookUp());

        // 4. join base_dic and cart_add
        Table cart_add_with_dic = tableEnv.sqlQuery("" +
                "SELECT " +
                    "ci.user_id, " +
                    "ci.sku_id, " +
                    "ci.cart_price, " +
                    "ci.sku_name, " +
                    "ci.is_checked, " +
                    "ci.create_time, " +
                    "ci.operate_time, " +
                    "ci.is_ordered, " +
                    "ci.order_time, " +
                    "ci.source_type source_type_code, " +
                    "bd.dic_name source_type_name ," +
                    "ci.source_id " +
                "FROM cart_add ci " +
                "JOIN base_dic FOR SYSTEM_TIME AS OF ci.pt as bd " +
                "ON ci.source_type = bd.dic_code " +
                "WHERE ci.source_type = bd.dic_code");

        tableEnv.createTemporaryView("cart_add_with_dic",cart_add_with_dic);

        // 5. create dwd_cart_add table
        tableEnv.executeSql("" +
                "create table dwd_cart_add( " +
                "    user_id  STRING,  " +
                "    sku_id  STRING,  " +
                "    cart_price  STRING,  " +
                "    sku_name  STRING,  " +
                "    is_checked  STRING,  " +
                "    create_time  STRING,  " +
                "    operate_time  STRING,  " +
                "    is_ordered  STRING,  " +
                "    order_time  STRING,  " +
                "    source_type_code  STRING,  " +
                "    source_type_name  STRING,  " +
                "    source_id  STRING " +
                ")" + MyKafkaUtil.getKafkaSinkDDL("dwd_cart_add"));

        // 6. write to dwd_cart_add
        cart_add_with_dic.executeInsert("dwd_cart_add");


    }
}
